//! 命名服务客户端
//!
//! 提供Nacos命名服务的完整功能实现，包括服务注册、发现、健康检查、负载均衡等
//! 支持多种服务选择策略（随机、轮询、权重）和服务变更监听机制
//!
//! # 功能特性
//! - 服务注册与注销
//! - 服务发现与实例列表获取
//! - 服务健康检查
//! - 服务变更监听与通知
//! - 缓存机制优化性能
//! - 多种负载均衡策略
//! - 重试机制保证高可用
//!
//! # 使用示例
//! ```rust
//! use nacos_client::{NacosClient, ClientConfig, ServiceInstance};
//! use std::collections::HashMap;
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
//!     let config = ClientConfig::new("127.0.0.1:8848");
//!     let client = NacosClient::new(config).await?;
//!     let naming_client = client.naming_client();
//!
//!     // 注册服务实例
//!     let instance = ServiceInstance::new(
//!         "test-service".to_string(),
//!         "DEFAULT_GROUP".to_string(),
//!         "127.0.0.1".to_string(),
//!         8080,
//!     );
//!     naming_client.register_instance("test-service", instance).await?;
//!
//!     // 获取服务实例列表
//!     let instances = naming_client.get_instances("test-service", true).await?;
//!     println!("发现 {} 个服务实例", instances.len());
//!
//!     Ok(())
//! }
//! ```

use std::collections::{HashMap, HashSet};
use std::hash::{Hash, Hasher};
use std::sync::Arc;
use tokio::sync::RwLock;
use tokio::time::{self, Duration};
use tracing::{error, info};

use nacos_core::ServiceInstance;
use crate::error::ClientResult;
use crate::retry::{RetryStrategy, RetryStrategyFactory};
use crate::NacosClientApi;

/// 命名服务客户端
///
/// 提供Nacos命名服务的核心功能，包括服务注册、发现、监听等
///
/// # 设计特点
/// - 线程安全：所有方法都是线程安全的
/// - 缓存优化：内置缓存减少网络请求
/// - 事件驱动：支持服务变更监听
/// - 高可用：内置重试和故障转移机制
#[derive(Clone)]
pub struct NamingClient {
    /// 关联的Nacos客户端
    client: Arc<dyn NacosClientApi>,
    /// 命名空间
    namespace: String,
    /// 重试策略
    retry_strategy: RetryStrategy,
    /// 服务实例缓存，key为服务名，value为实例列表
    cache: Arc<RwLock<HashMap<String, Vec<ServiceInstance>>>>,
    /// 服务监听器，key为服务名，value为监听器列表
    listeners: Arc<RwLock<HashMap<String, Vec<Box<dyn ServiceListener>>>>>,
    /// 需要发送心跳的实例
    heartbeat_instances: Arc<RwLock<HashSet<ServiceInstanceWrapper>>>,
    _phantom: std::marker::PhantomData<()>, // 占位符
}

#[async_trait::async_trait]
pub trait ServiceListener: Send + Sync + std::fmt::Debug {
    async fn on_service_changed(&self, service_name: &str, instances: &[ServiceInstance]);
    
    async fn on_instance_added(&self, service_name: &str, instance: &ServiceInstance);
    
    async fn on_instance_removed(&self, service_name: &str, instance: &ServiceInstance);
}

impl NamingClient {
    pub fn new(client: Arc<dyn NacosClientApi>, namespace: impl Into<String>) -> Self {
        let config = client.config();
        let retry_strategy = RetryStrategyFactory::create(config);
        
        let new_self = Self {
            client,
            namespace: namespace.into(),
            retry_strategy,
            cache: Arc::new(RwLock::new(HashMap::new())),
            listeners: Arc::new(RwLock::new(HashMap::new())),
            heartbeat_instances: Arc::new(RwLock::new(HashSet::new())),
            _phantom: std::marker::PhantomData,
        };

        new_self.start_heartbeat_task();
        new_self
    }

    /// 注册服务实例
    pub async fn register_instance(
        &self,
        service_name: &str,
        instance: ServiceInstance,
    ) -> ClientResult<()> {
        let client = self.client.clone();
        let service_name = service_name.to_string();
        let namespace = self.namespace.clone();
        
        let result = self.retry_strategy
            .execute(|| {
                let client = client.clone();
                let service_name = service_name.clone();
                let namespace = namespace.clone();
                let instance = instance.clone();
                
                async move {
                    client.register_instance(&namespace, &service_name, instance).await
                }
            })
            .await;

        if result.is_ok() {
            let mut instances = self.heartbeat_instances.write().await;
            instances.insert(instance.into());
        }
        result
    }

    /// 注销服务实例
    pub async fn deregister_instance(
        &self,
        service_name: &str,
        ip: &str,
        port: u16,
    ) -> ClientResult<()> {
        let client = self.client.clone();
        let service_name = service_name.to_string();
        let namespace = self.namespace.clone();
        
        let result = self.retry_strategy
            .execute(|| {
                let client = client.clone();
                let service_name = service_name.clone();
                let namespace = namespace.clone();
                let ip = ip.to_string();
                
                async move {
                    client.deregister_instance(&namespace, &service_name, &ip, port).await
                }
            })
            .await;

        if result.is_ok() {
            let mut instances = self.heartbeat_instances.write().await;
            instances.retain(|wrapper| {
                !(wrapper.0.service_name == service_name &&
                  wrapper.0.ip == ip &&
                  wrapper.0.port == port)
            });
        }
        result
    }

    /// 获取服务实例列表
    pub async fn get_instances(
        &self,
        service_name: &str,
        healthy_only: bool,
    ) -> ClientResult<Vec<ServiceInstance>> {
        let client = self.client.clone();
        let service_name = service_name.to_string();
        let namespace = self.namespace.clone();
        
        let instances = self.retry_strategy
            .execute(|| {
                let client = client.clone();
                let service_name = service_name.clone();
                let namespace = namespace.clone();
                
                async move {
                    client.get_instances(&namespace, &service_name, healthy_only).await
                }
            })
            .await?;

        // 更新缓存
        let mut cache = self.cache.write().await;
        cache.insert(service_name.clone(), instances.clone());
        
        Ok(instances)
    }

    /// 获取服务实例列表（带缓存）
    pub async fn get_instances_cached(
        &self,
        service_name: &str,
        healthy_only: bool,
    ) -> ClientResult<Vec<ServiceInstance>> {
        let service_name_str = service_name.to_string();
        
        // 先检查缓存
        {
            let cache = self.cache.read().await;
            if let Some(instances) = cache.get(&service_name_str) {
                if healthy_only {
                    return Ok(instances
                        .iter()
                        .filter(|instance| instance.healthy)
                        .cloned()
                        .collect());
                } else {
                    return Ok(instances.clone());
                }
            }
        }
        
        // 缓存未命中，从服务器获取
        self.get_instances(service_name, healthy_only).await
    }

    /// 获取所有服务列表
    pub async fn list_services(
        &self,
        page: u32,
        page_size: u32,
    ) -> ClientResult<Vec<String>> {
        let client = self.client.clone();
        let namespace = self.namespace.clone();
        
        self.retry_strategy
            .execute(|| {
                let client = client.clone();
                let namespace = namespace.clone();
                
                async move {
                    client.list_services(&namespace, page, page_size).await
                }
            })
            .await
    }

    /// 订阅服务变更
    pub async fn subscribe_service(
        &self,
        service_name: &str,
        listener: Box<dyn ServiceListener>,
    ) -> ClientResult<()> {
        let service_name = service_name.to_string();
        
        let mut listeners = self.listeners.write().await;
        listeners
            .entry(service_name)
            .or_insert_with(Vec::new)
            .push(listener);
        
        Ok(())
    }

    /// 取消订阅服务变更
    pub async fn unsubscribe_service(
        &self,
        service_name: &str,
    ) -> ClientResult<()> {
        let service_name = service_name.to_string();
        
        let mut listeners = self.listeners.write().await;
        listeners.remove(&service_name);
        
        Ok(())
    }

    /// 更新服务实例
    pub async fn update_instance(
        &self,
        service_name: &str,
        instance: ServiceInstance,
    ) -> ClientResult<()> {
        // 先注销再注册
        self.deregister_instance(service_name, &instance.ip, instance.port).await?;
        self.register_instance(service_name, instance).await
    }

    /// 获取服务实例数量
    pub async fn get_instance_count(
        &self,
        service_name: &str,
        healthy_only: bool,
    ) -> ClientResult<usize> {
        let instances = self.get_instances(service_name, healthy_only).await?;
        Ok(instances.len())
    }

    /// 检查服务是否存在
    pub async fn service_exists(&self, service_name: &str) -> ClientResult<bool> {
        let services = self.list_services(1, 100).await?;
        Ok(services.contains(&service_name.to_string()))
    }

    /// 获取健康实例
    pub async fn get_healthy_instances(
        &self,
        service_name: &str,
    ) -> ClientResult<Vec<ServiceInstance>> {
        self.get_instances(service_name, true).await
    }

    /// 获取命名空间
    pub fn namespace(&self) -> &str {
        &self.namespace
    }

    /// 强制刷新缓存
    pub async fn refresh_cache(&self, service_name: &str) -> ClientResult<()> {
        let instances = self.get_instances(service_name, false).await?;
        
        let mut cache = self.cache.write().await;
        cache.insert(service_name.to_string(), instances);
        
        Ok(())
    }

    /// 获取缓存的服务实例
    pub async fn get_cached_instances(&self, service_name: &str) -> Option<Vec<ServiceInstance>> {
        let cache = self.cache.read().await;
        cache.get(service_name).cloned()
    }

    /// 清除缓存
    pub async fn clear_cache(&self) {
        let mut cache = self.cache.write().await;
        cache.clear();
    }

    /// 通知监听器服务变更
    async fn notify_listeners(
        &self,
        service_name: &str,
        instances: &[ServiceInstance],
    ) {
        let listeners = self.listeners.read().await;
        
        if let Some(service_listeners) = listeners.get(service_name) {
            for listener in service_listeners {
                listener.on_service_changed(service_name, instances).await;
            }
        }
    }

    fn start_heartbeat_task(&self) {
        let client = self.client.clone();
        let namespace = self.namespace.clone();
        let heartbeat_instances = self.heartbeat_instances.clone();

        tokio::spawn(async move {
            let mut interval = time::interval(Duration::from_secs(5));
            loop {
                interval.tick().await;
                let instances = heartbeat_instances.read().await;
                for wrapper in instances.iter() {
                    let instance = wrapper.0.clone();
                    let service_name = instance.service_name.clone();
                    if let Err(e) = client.send_heartbeat(&namespace, &service_name, instance).await {
                        error!("Failed to send heartbeat: {}", e);
                    }
                }
            }
        });
    }
}

#[derive(Clone, Debug)]
struct ServiceInstanceWrapper(ServiceInstance);

impl Eq for ServiceInstanceWrapper {}

impl PartialEq for ServiceInstanceWrapper {
    fn eq(&self, other: &Self) -> bool {
        self.0.service_name == other.0.service_name &&
        self.0.group_name == other.0.group_name &&
        self.0.ip == other.0.ip &&
        self.0.port == other.0.port
    }
}

impl Hash for ServiceInstanceWrapper {
    fn hash<H: Hasher>(&self, state: &mut H) {
        self.0.service_name.hash(state);
        self.0.group_name.hash(state);
        self.0.ip.hash(state);
        self.0.port.hash(state);
    }
}

impl From<ServiceInstance> for ServiceInstanceWrapper {
    fn from(instance: ServiceInstance) -> Self {
        Self(instance)
    }
}

/// 默认服务监听器
#[derive(Debug)]
pub struct DefaultServiceListener;

#[async_trait::async_trait]
impl ServiceListener for DefaultServiceListener {
    async fn on_service_changed(&self, service_name: &str, instances: &[ServiceInstance]) {
        info!("Service {} changed, instances: {}", service_name, instances.len());
    }

    async fn on_instance_added(&self, service_name: &str, instance: &ServiceInstance) {
        info!("Instance added to {}: {}:{}", service_name, instance.ip, instance.port);
    }

    async fn on_instance_removed(&self, service_name: &str, instance: &ServiceInstance) {
        info!("Instance removed from {}: {}:{}", service_name, instance.ip, instance.port);
    }
}

/// 服务选择器
///
/// 用于从多个服务实例中选择一个最优实例的算法接口
/// 支持多种负载均衡策略，提高系统的可用性和性能
#[async_trait::async_trait]
pub trait ServiceSelector: Send + Sync {
    /// 从实例列表中选择一个实例
    ///
    /// # 参数
    /// * `instances` - 服务实例列表
    ///
    /// # 返回值
    /// 返回选中的服务实例，如果列表为空则返回None
    fn select<'a>(&self, instances: &'a [ServiceInstance]) -> Option<&'a ServiceInstance>;
}

/// 随机服务选择器
///
/// 使用随机算法从可用的服务实例中选择一个实例
/// 适用于实例性能相近的场景
pub struct RandomSelector;

impl ServiceSelector for RandomSelector {
    fn select<'a>(&self, instances: &'a [ServiceInstance]) -> Option<&'a ServiceInstance> {
        if instances.is_empty() {
            return None;
        }
        
        use rand::Rng;
        let mut rng = rand::thread_rng();
        let index = rng.gen_range(0..instances.len());
        Some(&instances[index])
    }
}

/// 轮询服务选择器
///
/// 使用轮询算法依次选择每个服务实例
/// 确保每个实例都能均匀接收到请求
pub struct RoundRobinSelector {
    /// 当前选择的索引位置
    index: Arc<RwLock<usize>>,
}

impl RoundRobinSelector {
    /// 创建新的轮询选择器
    pub fn new() -> Self {
        Self {
            index: Arc::new(RwLock::new(0)),
        }
    }
}

impl ServiceSelector for RoundRobinSelector {
    fn select<'a>(&self, instances: &'a [ServiceInstance]) -> Option<&'a ServiceInstance> {
        if instances.is_empty() {
            return None;
        }
        
        let mut index = self.index.try_write().unwrap_or_else(|_| self.index.blocking_write());
        let instance = &instances[*index];
        *index = (*index + 1) % instances.len();
        Some(instance)
    }
}

impl Default for RoundRobinSelector {
    fn default() -> Self {
        Self::new()
    }
}

/// 权重服务选择器
///
/// 根据服务实例的权重进行加权随机选择
/// 权重高的实例被选中的概率更大
///
/// # 权重计算
/// 每个实例被选中的概率 = 实例权重 / 所有实例权重之和
pub struct WeightedSelector;

impl ServiceSelector for WeightedSelector {
    fn select<'a>(&self, instances: &'a [ServiceInstance]) -> Option<&'a ServiceInstance> {
        if instances.is_empty() {
            return None;
        }
        
        // 计算总权重
        let total_weight: f64 = instances.iter().map(|i| i.weight as f64).sum();
        if total_weight <= 0.0 {
            // 如果所有权重都小于等于0，返回第一个实例
            return Some(&instances[0]);
        }
        
        // 加权随机选择
        use rand::Rng;
        let mut rng = rand::thread_rng();
        let random_weight = rng.gen_range(0.0..total_weight);
        
        let mut current_weight = 0.0;
        for instance in instances {
            current_weight += instance.weight as f64;
            if random_weight <= current_weight {
                return Some(instance);
            }
        }
        
        // 兜底返回第一个实例
        Some(&instances[0])
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::client::MockNacosClient;
    use crate::NacosClientApi;

    #[tokio::test]
    async fn test_naming_client_creation() {
        let mock_client = Arc::new(MockNacosClient::new());
        let naming_client = NamingClient::new(mock_client as Arc<dyn NacosClientApi>, "test-namespace");
        
        assert_eq!(naming_client.namespace(), "test-namespace");
    }

    #[tokio::test]
    async fn test_register_instance() {
        let mock_client = Arc::new(MockNacosClient::new());
        let naming_client = NamingClient::new(mock_client as Arc<dyn NacosClientApi>, "test-namespace");
        
        let instance = ServiceInstance::new(
            "test-service".to_string(),
            "DEFAULT_GROUP".to_string(),
            "127.0.0.1".to_string(),
            8080,
        );
        
        let result = naming_client.register_instance("test-service", instance).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_get_instances() {
        let mock_client = Arc::new(MockNacosClient::new());
        let naming_client = NamingClient::new(mock_client as Arc<dyn NacosClientApi>, "test-namespace");
        
        let instances = naming_client.get_instances("test-service", true).await.unwrap();
        assert!(!instances.is_empty());
    }

    #[tokio::test]
    async fn test_service_selector_random() {
        let selector = RandomSelector;
        let instances = vec![
            ServiceInstance {
                ip: "127.0.0.1".to_string(),
                port: 8080,
                ..Default::default()
            },
            ServiceInstance {
                ip: "127.0.0.1".to_string(),
                port: 8081,
                ..Default::default()
            },
        ];
        
        let selected = selector.select(&instances);
        assert!(selected.is_some());
    }

    #[tokio::test]
    async fn test_service_selector_round_robin() {
        let selector = RoundRobinSelector::new();
        let instances = vec![
            ServiceInstance {
                ip: "10.0.0.1".to_string(),
                port: 8080,
                ..Default::default()
            },
            ServiceInstance {
                ip: "10.0.0.2".to_string(),
                port: 8081,
                ..Default::default()
            },
        ];
        
        let selected1 = selector.select(&instances);
        let selected2 = selector.select(&instances);
        
        assert!(selected1.is_some());
        assert!(selected2.is_some());
        assert_ne!(selected1.unwrap().port, selected2.unwrap().port);
    }

    #[tokio::test]
    async fn test_service_selector_weighted() {
        let selector = WeightedSelector;
        let instances = vec![
            ServiceInstance {
                ip: "10.0.0.1".to_string(),
                port: 8080,
                weight: 0.3,
                ..Default::default()
            },
            ServiceInstance {
                ip: "10.0.0.2".to_string(),
                port: 8081,
                weight: 0.7,
                ..Default::default()
            },
        ];
        
        let selected = selector.select(&instances);
        assert!(selected.is_some());
    }

    #[tokio::test]
    async fn test_cache_functionality() {
        let mock_client = Arc::new(MockNacosClient::new());
        let naming_client = NamingClient::new(mock_client as Arc<dyn NacosClientApi>, "test-namespace");
        
        let instances = naming_client.get_instances("test-service", true).await.unwrap();
        
        // 检查缓存
        let cached = naming_client.get_cached_instances("test-service").await;
        assert!(cached.is_some());
        assert_eq!(cached.unwrap().len(), instances.len());
    }

    #[tokio::test]
    async fn test_service_exists() {
        let mock_client = Arc::new(MockNacosClient::new());
        let naming_client = NamingClient::new(mock_client as Arc<dyn NacosClientApi>, "test-namespace");
        
        let exists = naming_client.service_exists("test-service").await.unwrap();
        assert!(exists);
    }

    #[tokio::test]
    async fn test_subscribe_service() {
        let mock_client = Arc::new(MockNacosClient::new());
        let naming_client = NamingClient::new(mock_client as Arc<dyn NacosClientApi>, "test-namespace");
        
        let listener = Box::new(DefaultServiceListener);
        let result = naming_client.subscribe_service("test-service", listener).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_unsubscribe_service() {
        let mock_client = Arc::new(MockNacosClient::new());
        let naming_client = NamingClient::new(mock_client as Arc<dyn NacosClientApi>, "test-namespace");
        
        let result = naming_client.unsubscribe_service("test-service").await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_clear_cache() {
        let mock_client = Arc::new(MockNacosClient::new());
        let naming_client = NamingClient::new(mock_client as Arc<dyn NacosClientApi>, "test-namespace");
        
        naming_client.clear_cache().await;
        
        let cached = naming_client.get_cached_instances("test-service").await;
        assert!(cached.is_none());
    }

    #[tokio::test]
    async fn test_get_healthy_instances() {
        let mock_client = Arc::new(MockNacosClient::new());
        let naming_client = NamingClient::new(mock_client as Arc<dyn NacosClientApi>, "test-namespace");
        
        let instances = naming_client.get_healthy_instances("test-service").await.unwrap();
        assert!(!instances.is_empty());
    }

    #[tokio::test]
    async fn test_update_instance() {
        let mock_client = Arc::new(MockNacosClient::new());
        let naming_client = NamingClient::new(mock_client as Arc<dyn NacosClientApi>, "test-namespace");
        
        let instance = ServiceInstance::new(
            "test-service".to_string(),
            "DEFAULT_GROUP".to_string(),
            "127.0.0.1".to_string(),
            8080,
        );
        
        let result = naming_client.update_instance("test-service", instance).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_empty_instances() {
        let mock_client = Arc::new(MockNacosClient::new());
        let naming_client = NamingClient::new(mock_client as Arc<dyn NacosClientApi>, "test-namespace");
        
        let instances = naming_client.get_instances("empty-service", true).await.unwrap();
        assert!(instances.is_empty());
    }
}